package com.mygame.gateway.server.handler;

import com.mygame.common.codec.DataCodecKit;
import com.mygame.gateway.server.queue.NatsQueue;
//import org.apache.kafka.clients.producer.ProducerRecord;
import com.mygame.common.protobuf.ExternalMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.mygame.common.cloud.PlayerServiceInstance;
import com.mygame.common.utils.JWTUtil.TokenBody;
import com.mygame.common.utils.NettyUtils;
import com.mygame.common.utils.TopicUtil;
import com.mygame.gateway.server.GatewayServerConfig;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;

public class DispatchGameMessageHandler extends ChannelInboundHandlerAdapter {
    private PlayerServiceInstance playerServiceInstance;// 注入业务服务管理类，从这里获取负载均衡的服务器信息
    private GatewayServerConfig gatewayConfig; // 注入游戏网关服务配置信息。
    private TokenBody tokenBody;

    private static Logger logger = LoggerFactory.getLogger(DispatchGameMessageHandler.class);

    public DispatchGameMessageHandler(PlayerServiceInstance playerServiceInstance, GatewayServerConfig gatewayServerConfig) {
        this.playerServiceInstance = playerServiceInstance;
        this.gatewayConfig = gatewayServerConfig;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        ExternalMessage gameMessagePackage = (ExternalMessage) msg;
        int serviceId = gameMessagePackage.getServiceId();
        if (tokenBody == null) {// 如果首次通信，获取验证信息
            ConfirmHandler confirmHandler = (ConfirmHandler) ctx.channel().pipeline().get("ConfirmHandler");
            tokenBody = confirmHandler.getTokenBody();
        }
        String clientIp = NettyUtils.getRemoteIP(ctx.channel());
        dispatchMessage(ctx.executor(), playerServiceInstance, tokenBody.getPlayerId(), serviceId, clientIp, gameMessagePackage, gatewayConfig);
    }

    /**
     * 网关分发消息到业务服
     */
    public static void dispatchMessage(EventExecutor executor, PlayerServiceInstance playerServiceInstance, long playerId, int serviceId, String clientIp, ExternalMessage gameMessagePackage, GatewayServerConfig gatewayServerConfig) {
        Promise<Integer> promise = new DefaultPromise<>(executor);
        playerServiceInstance.selectServerId(playerId, serviceId, promise).addListener(new GenericFutureListener<Future<Integer>>() {

            @Override
            public void operationComplete(Future<Integer> future) throws Exception {
                if (future.isSuccess()) {
                    Integer toServerId = future.get();
                    gameMessagePackage.setToServerId(toServerId);
                    gameMessagePackage.setFromServerId(gatewayServerConfig.getServerId());
//                    gameMessagePackage.getAttribute().setClientIp(clientIp);
                    gameMessagePackage.setPlayerId(playerId);
                    String topic = TopicUtil.generateTopic(gatewayServerConfig.getBusinessGameMessageTopic(), toServerId);// 动态创建与业务服务交互的消息总线Topic
                    byte[] value = DataCodecKit.encode(gameMessagePackage);// 向消息总线服务发布客户端请求消息。
                    NatsQueue.INSTANCE.getSc().publish(topic, value);
                    logger.debug("发送到{}消息成功->{}", gameMessagePackage.getMessageId());
                } else {
                    logger.error("消息发送失败", future.cause());
                }
            }
        });
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
        logger.error("服务器异常，连接{}断开", ctx.channel().id().asShortText(), cause);
    }
}
